package cn.ipanel.bigdata.dw.realtime

import cn.ipanel.bigdata.boot.source.genre.Phoenix
import cn.ipanel.bigdata.dw.dim.phoenix.saltBuckets
import cn.ipanel.bigdata.dw.BaseColumns
import cn.ipanel.bigdata.dw.dws.HBASE_NAMESPACE
import org.apache.spark.sql.{Dataset, Row}

/**
 * @author lzz
 * @environment IntelliJ IDEA 2020.3.1
 * @projectName bigdata_shanxi
 * @date 2023/04/04 11:03
 * @description:
 */
protected[realtime] object Network extends Phoenix(HBASE_NAMESPACE, "t_device_network")  with BaseColumns{

  final val F_EXCELLENT_NUM     : String = "f_excellent_num"
  final val F_GOOD_NUM          : String = "f_good_num"
  final val F_BAD_NUM           : String = "f_bad_num"

  override def getTBColumns: Seq[String] = {
    Seq(F_PERIOD_TIME, F_REGION, F_DEVICE_COUNT, F_EXCELLENT_NUM, F_GOOD_NUM, F_BAD_NUM)
  }

  override def buildTable: String = {
    s"""
       |CREATE TABLE IF NOT EXISTS $getDBName.$getTBName (
       |$F_PERIOD_TIME BIGINT NOT NULL
       |, $F_REGION BIGINT NOT NULL
       |, $F_DEVICE_COUNT INTEGER
       |, $F_EXCELLENT_NUM BIGINT
       |, $F_GOOD_NUM BIGINT
       |, $F_BAD_NUM BIGINT
       | CONSTRAINT PK PRIMARY KEY
       |  ($F_PERIOD_TIME, $F_REGION)) SALT_BUCKETS = $saltBuckets
       |""".stripMargin
  }

  override def emptyTable: Dataset[NetworkTable] = {
    import IMPLICITS._
    spark.createDataset(spark.sparkContext.emptyRDD[NetworkTable])
  }

  case class NetworkTable(var f_period_time: Long
                          , var f_region: Long
                          , var f_device_count: Int
                          , var f_excellent_num: Long
                          , var f_good_num: Long
                          , var f_bad_num: Long
                                )

  object NetworkTable {

    def apply(row: Row): NetworkTable = {
      NetworkTable(row.getAs[Long](F_PERIOD_TIME)
        , row.getAs[Long](F_REGION)
        , row.getAs[Int](F_DEVICE_COUNT)
        , row.getAs[Long](F_EXCELLENT_NUM)
        , row.getAs[Long](F_GOOD_NUM)
        , row.getAs[Long](F_BAD_NUM)
      )
    }
  }

}
